package com.letv.whatslive.mongo.dao;

import com.google.common.collect.Lists;
import com.letv.whatslive.common.utils.CacheConstants;
import com.letv.whatslive.common.utils.Constants;
import com.letv.whatslive.common.utils.DateUtils;
import com.letv.whatslive.common.utils.ObjectUtils;
import com.letv.whatslive.model.Program;
import com.letv.whatslive.model.convert.ProgramConvert;
import com.letv.whatslive.model.utils.ObjUtils;
import com.letv.whatslive.mongo.BaseDAO;
import com.letv.whatslive.redis.JedisDAO;
import com.mongodb.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import java.util.*;
import java.util.regex.Pattern;

/**
 * 直播Collection操作DAO
 * Created by gaoshan on 15-7-13.
 */
@Repository
public class ProgramDAO extends BaseDAO {

    private final static Logger logger = LoggerFactory.getLogger(ProgramDAO.class);

    protected String collectionName = "program";

    @Autowired
    protected JedisDAO jedisDAO;

    @Override
    protected void init() {
        super.init(this.collectionName);
    }

    @Override
    protected long getAutoIncrementId() {
        return idGenerate.generateId(this.collectionName);
    }

    public long insertProgram(Program program) {
        DBObject obj = ProgramConvert.castProgramToDBObject(program);
        DBRef userRef = new DBRef("user", program.getUserId());
        obj.put("user", userRef);
        return this.insert(obj);
    }

    public List<Program> getAllProgram(Integer start, Integer limit, Map<String, Object> params) {

        // // TODO 时移播放
//        params.put("status",1);
        DBObject order = new BasicDBObject();
        order.put("pType", 1);
        order.put("priority", -1);
        order.put("createTime", -1);
        List<DBObject> programList = selectAll(start, limit, params, order);
        List<Program> result = Lists.newArrayList();
        for (DBObject obj : programList) {
            Program program = ProgramConvert.castDBObjectToProgram(obj);

            // 设定视频参与信息
            setInvolInfor(program);

            result.add(program);
        }
        return result;
    }

    public List<DBObject> getUpdateProgramListByIds(BasicDBList userIds, Long timestamp, Integer start, Integer limit) {
        List<DBObject> result = Lists.newArrayList();
        try {
            DBObject queryCondition = new BasicDBObject();
            queryCondition.put("userId", new BasicDBObject("$in", userIds));
            queryCondition.put("createTime", new BasicDBObject("$gt", timestamp));

            // 时移播放
            BasicDBList statusList = new BasicDBList();
            statusList.add(-1);
            statusList.add(1);
            queryCondition.put("status", new BasicDBObject("$in", statusList));

            DBObject order = new BasicDBObject();
            order.put("pType", 1);
            DBCursor cur = dbCollection.find(queryCondition).sort(order).skip((start - 1) * limit).limit(limit);
            while (cur.hasNext()) {
                result.add(cur.next());

            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public List<Program> getProgramListByIds(BasicDBList userIds, Map<String, Object> params, Integer start, Integer limit) {
        List<Program> result = Lists.newArrayList();
        try {
            DBObject queryCondition = new BasicDBObject();
            queryCondition.put("userId", new BasicDBObject("$in", userIds));
            if (params != null) {
                Iterator iter = params.entrySet().iterator();
                while (iter.hasNext()) {
                    Map.Entry entry = (Map.Entry) iter.next();
                    queryCondition.put(ObjectUtils.toString(entry.getKey()), entry.getValue());
                }
            }
            DBObject order = new BasicDBObject();
            order.put("pType", 1);
            order.put("startTime", -1);
            DBCursor cur = dbCollection.find(queryCondition).sort(order).skip((start - 1) * limit).limit(limit);
            while (cur.hasNext()) {

                Program program = ProgramConvert.castDBObjectToProgram(cur.next());

                // 设定视频参与信息
                setInvolInfor(program);

                result.add(program);

            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    /**
     * 查询视频列表
     *
     * @param orders 查询的排序条件
     * @param params 模糊查询的查询条件
     * @param start  查询条件的开始分页数
     * @param limit  查询条件的分页每页条数
     * @return
     */
    public List<Program> getProgramListByParams(Map<String, Object> orders, Map<String, Object> params, Integer start, Integer limit) {

        List<Program> result = Lists.newArrayList();
        if (limit == null) {
            limit = DEFAULT_NUM;
        }
        try {
            //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
            BasicDBObject query = getBasicDBObjectByParams(params);
            BasicDBObject order = getOrderParams(orders);
            Integer skip = start == null ? 0 : start;
            DBCursor cur = this.dbCollection.find(query).skip(skip)
                    .sort(order).limit(limit);//按照视频类型和创建时间倒序
            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                setValueToProgram(program);
                result.add(program);
            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public List<Long> getProgramIdsByParams(Map<String, Object> params, Map<String, Object> orders, String... readPreference) {

        List<Long> result = Lists.newArrayList();
        try {
            //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
            BasicDBObject query = getBasicDBObjectByParams(params);
            BasicDBObject order = getOrderParams(orders);

            DBCursor cur = null;

            if (readPreference != null && readPreference.length > 0
                    && Constants.MONGO_READPREFERENCE_PRIMARY.equals(readPreference[0])) {

                cur = this.dbCollectionOnPrimary.find(query).sort(order);
            } else {

                cur = this.dbCollection.find(query).sort(order);
            }

            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                result.add(ObjUtils.toLong(dbObject.get("_id")));
            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    /**
     * 根据多个Id的字符串查询直播信息
     *
     * @param ids            多个以英文逗号分隔
     * @param readPreference 读库指定
     * @return
     */
    public List<Program> getProgramListByIds(String ids, String... readPreference) {
        List<Program> result = Lists.newArrayList();
        try {
            BasicDBObject query = new BasicDBObject();// 新建查询基类对象 dbo
            String[] idList = ids.split(",");
            BasicDBList values = new BasicDBList();
            for (String id : idList) {
                values.add(ObjectUtils.toLong(id));
            }
            query.put("_id", new BasicDBObject("$in", values));
            DBCursor cur = null;

            if (readPreference != null && readPreference.length > 0
                    && Constants.MONGO_READPREFERENCE_PRIMARY.equals(readPreference[0])) {

                cur = this.dbCollectionOnPrimary.find(query);
            } else {

                cur = this.dbCollection.find(query);
            }

            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                setValueToProgram(program);
                result.add(program);
            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    /**
     * 根据多个Id查询直播信息，为客户端使用，非cms
     *
     * @param idList
     * @return
     */
    public List<Program> getProgramListByIds(List<Long> idList) {
        List<Program> result = Lists.newArrayList();
        try {
            BasicDBObject query = new BasicDBObject();// 新建查询基类对象 dbo
            BasicDBList values = new BasicDBList();
            for (Long id : idList) {
                values.add(id);
            }
            query.put("_id", new BasicDBObject("$in", values));
            DBCursor cur = dbCollection.find(query);
            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                setValueToProgram(program);
                result.add(program);
            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }


    /**
     * 查询直播的记录数
     *
     * @param params 查询条件参数
     * @return 满足条件的文档记录数
     */
    public Long countProgramByParams(Map params, String... readPreference) {
        //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
        BasicDBObject query = getBasicDBObjectByParams(params);

        Long count = null;

        if (readPreference != null && readPreference.length > 0
                && Constants.MONGO_READPREFERENCE_PRIMARY.equals(readPreference[0])) {

            count = this.dbCollectionOnPrimary.count(query);
        } else {

            count = this.dbCollection.count(query);
        }

        return count;
    }

    /**
     * 根据直播ID查询直播信息
     *
     * @param id
     * @return
     */
    public Program getProgramById(Long id, String... readPreference) {
        DBObject dbObject = new BasicDBObject();
        dbObject.put("_id", id);

        DBObject result = null;
        if (readPreference != null && readPreference.length > 0
                && Constants.MONGO_READPREFERENCE_PRIMARY.equals(readPreference[0])) {

            result = this.dbCollectionOnPrimary.findOne(dbObject);
        } else {

            result = dbCollection.findOne(dbObject);
        }

        if (result == null) {
            return null;
        }
        Program program = ProgramConvert.castDBObjectToProgram(result);
        setValueToProgram(program);
        return program;
    }

    /**
     * 更新直播信息，新增内容到数组，避免重复
     *
     * @param addValue
     * @param programId
     */
    public void updateProgramAddToSet(Map<String, Object> addValue, Long programId) {
        DBObject queryCondition = new BasicDBObject();
        queryCondition.put("_id", programId); // 用来确定要修改的文档
        DBObject updateValue = new BasicDBObject("$addToSet", addValue);
        this.dbCollection.update(queryCondition, updateValue, true, false);
    }

    /**
     * 更新直播信息
     *
     * @param program
     */
    public void updateProgram(Program program) {
        DBObject queryCondition = new BasicDBObject();
        Long pid = program.getId();
        queryCondition.put("_id", pid); // 用来确定要修改的文档
        BasicDBObject setValue = (BasicDBObject) ProgramConvert.castProgramToDBObject(program);
        //剔除主键修改
        if (setValue.containsField("_id")) {
            setValue.removeField("_id");
        }
        DBRef userRef = new DBRef(this.dbCollection.getDB(), "user", program.getUserId());
        setValue.put("user", userRef);
        DBObject updateValue = new BasicDBObject("$set", setValue);
        this.dbCollection.update(queryCondition, updateValue, true, false);
    }

    /**
     * 根据直播ID删除直播
     *
     * @param pid
     */
    public void deleteProgram(Long pid) {
        BasicDBObject query = new BasicDBObject();
        query.put("_id", pid);
        this.dbCollection.remove(query);
    }

    /**
     * 根据查询参数Map获取查询条件对象
     *
     * @param params 查询参数Map
     * @return
     */
    private BasicDBObject getBasicDBObjectByParams(Map params) {
        //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
        BasicDBObject query = new BasicDBObject();// 新建查询基类对象 dbo
        if (params != null && params.size() > 0) {
            if (params.get("id") != null) {
                query.put("_id", ObjectUtils.toLong(params.get("id")));
            }
            if (params.get("userId") != null) {
                query.put("userId", ObjectUtils.toLong(params.get("userId")));
            }
            if (params.get("pType") != null) {
                String pType = ObjectUtils.toString(params.get("pType"));
                if (pType.length() > 2) {
                    BasicDBList values = new BasicDBList();
                    String[] pTypeArray = pType.split(",");
                    for (String type : pTypeArray) {
                        values.add(ObjectUtils.toInteger(type));
                    }
                    query.put("pType", new BasicDBObject("$in", values));
                } else {
                    query.put("pType", ObjectUtils.toInteger(pType));
                }
            }
            if (params.get("pName") != null) {
                BasicDBList values = new BasicDBList();
                Pattern pattern = Pattern.compile("^.*" + params.get("pName") + ".*$", Pattern.CASE_INSENSITIVE);
                values.add(new BasicDBObject("_id", ObjectUtils.toLong(params.get("pName"))));
                values.add(new BasicDBObject("pName", pattern));
                query.put("$or", values);
            }
            if (params.get("pFullName") != null) {
                query.put("pName", params.get("pFullName"));
            }
            if (params.get("timeShift") != null) {
                BasicDBList statusList = new BasicDBList();
                statusList.add(-1);
                statusList.add(1);
                query.put("status", new BasicDBObject("$in", statusList));
            } else {
                if (params.get("status") != null) {
                    String status = ObjectUtils.toString(params.get("status"));
                    if (status.length() > 2) {
                        BasicDBList values = new BasicDBList();
                        String[] pTypeArray = status.split(",");
                        for (String type : pTypeArray) {
                            values.add(ObjectUtils.toInteger(type));
                        }
                        query.put("status", new BasicDBObject("$in", values));
                    } else {
                        query.put("status", ObjectUtils.toInteger(status));
                    }
                }
            }
            if (params.get("reportNum") != null) {
                query.put("reportNum", new BasicDBObject("$gt", 0L));
            }
            if (params.get("startTime_start") != null && params.get("startTime_end") != null) {
                query.put("startTime", new BasicDBObject("$gt",
                        ObjectUtils.toLong(params.get("startTime_start"))).append("$lte", ObjectUtils.toLong(params.get("startTime_end"))));

            }
            if (params.get("endTime") != null) {
                query.put("startTime", new BasicDBObject("$lte",
                        ObjectUtils.toLong(params.get("endTime"))));
            }
            if (params.get("isCarousel") != null) {
                query.put("isCarousel", 1);
            }
            if (params.get("isNotCarousel") != null) {
                query.put("isCarousel", new BasicDBObject("$ne", 1));
            }
            if (params.get("activityId") != null) {
                query.put("activityId", ObjectUtils.toString(params.get("activityId")));
            }
            if (params.get("liveId") != null) {
                BasicDBList liveIdList = new BasicDBList();
                BasicDBObject dbObject = new BasicDBObject();
                dbObject.put("liveId", new BasicDBObject("$exists", params.get("liveId")));
                liveIdList.add(dbObject);
                liveIdList.add(new BasicDBObject("liveId", ""));
                query.put("$or", liveIdList);
            }
            if (params.get("exists_assorts") != null) {
                query.put("assorts.id", new BasicDBObject("$exists", true));
            }
            if (params.get("assorts.id") != null) {
                query.put("assorts.id", ObjectUtils.toLong(params.get("assorts.id")));
            }
            if (params.get("activity.id") != null) {
                query.put("activity.id", ObjectUtils.toLong(params.get("activity.id")));
            }
            if (params.get("notEquals") != null) {
                query.put("_id", new BasicDBObject("$ne", params.get("notEquals")));
            }
            if (params.get("isShow") != null) {
                query.put("isShow", ObjectUtils.toInteger(params.get("isShow")));
            }
            if (params.get("from") != null) {
                query.put("from", ObjectUtils.toInteger(params.get("from")));
            }
            if (params.get("realCommentNum") != null) {
                query.put("realCommentNum", ObjectUtils.toLong(params.get("realCommentNum")));
            }
            if (params.get("isPrivacy") != null) {
                if (1 == params.get("isPrivacy")) {
                    query.put("isPrivacy", 1);
                } else {
                    query.put("isPrivacy", new BasicDBObject("$ne", 1));
                }
            }
            if (params.get("idOrTitle") != null) {
                BasicDBList values = new BasicDBList();
                Pattern pattern = Pattern.compile("^.*" + params.get("idOrTitle") + ".*$", Pattern.CASE_INSENSITIVE);
                values.add(new BasicDBObject("_id", ObjectUtils.toLong(params.get("idOrTitle"))));
                values.add(new BasicDBObject("pName", pattern));
                values.add(new BasicDBObject("primaryPName", pattern));
                query.put("$or", values);
            }
            if (params.get("videoUploadRole") != null) {
                query.put("videoUploadRole", ObjectUtils.toInteger(params.get("videoUploadRole")));
            }
            if (params.get("sourceType") != null) {
                query.put("sourceType", ObjectUtils.toInteger(params.get("sourceType")));
            }
            if (params.get("userIds") != null) {
                List<Long> userIds = (List<Long>) params.get("userIds");
                BasicDBList values = new BasicDBList();
                if (userIds != null && userIds.size() > 0) {
                    for (Long userid : userIds) {
                        values.add(ObjectUtils.toLong(userid));
                    }
                    query.put("userId", new BasicDBObject("$in", values));
                }
            }
            if (params.get("appId") != null) {
                query.put("appId", ObjectUtils.toString(params.get("appId")));
            }
            if (params.get("isNotGoPro") != null) {
                query.put("source", new BasicDBObject("$ne", "GoPro"));
            }
            if (params.get("userRole") != null) {
                if (params.get("userRole") != null) {
                    String userRole = ObjectUtils.toString(params.get("userRole"));
                    if (userRole.length() > 2) {
                        BasicDBList values = new BasicDBList();
                        String[] userRoleArray = userRole.split(",");
                        for (String role : userRoleArray) {
                            values.add(ObjectUtils.toInteger(role));
                        }
                        query.put("userRole", new BasicDBObject("$in", values));
                    } else {
                        query.put("userRole", ObjectUtils.toInteger(userRole));
                    }
                }
            }
            if (params.get("appId") != null) {
                query.put("appId", ObjectUtils.toString(params.get("appId")));
            }
        }
        return query;
    }

    /**
     * 获取排序的BasicDBObject对象
     *
     * @param orders
     * @return
     */
    private BasicDBObject getOrderParams(Map<String, Object> orders) {
        BasicDBObject order = new BasicDBObject();
        if (null != orders && orders.size() > 0) {
            for (String key : orders.keySet()) {
                order.append(key, Integer.valueOf(String.valueOf(orders.get(key))));
            }
        } else {
            //默认按照创建时间排序
            order.append("createTime", -1);
        }
        return order;
    }

    /**
     * 设置直播信息中的特殊值
     *
     * @param program
     */
    private void setValueToProgram(Program program) {
        program.setStartTimeStr(program.getStartTime() != null ?
                DateUtils.long2YMDHMS(program.getStartTime()) : "");
        program.setEndTimeStr(program.getEndTime() != null ?
                DateUtils.long2YMDHMS(program.getEndTime()) : "");
        program.setCreateTimeStr(program.getCreateTime() != null ?
                DateUtils.long2YMDHMS(program.getCreateTime()) : "");
    }

    /**
     * 根据指定查询条件，分组条件获得指定项目的合计值。
     *
     * @param queryParams   查询条件参数
     * @param groupParams   分组条件参数
     * @param sumColumnName 计算合计值得列名称
     * @return 以组为单位的合计值
     */
    public Map<String, Long> getTotalNumByParams(Map queryParams, List<String> groupParams, String sumColumnName) {

        // 合计值
        Map<String, Long> totalNumMap = new HashMap<String, Long>();
        List<DBObject> pipeline = new ArrayList<DBObject>();

        // 检索条件
        BasicDBObject matchDBObj = new BasicDBObject();
        BasicDBObject matchCondition = new BasicDBObject();
        if (queryParams.get("pType") != null) {

            String pType = ObjectUtils.toString(queryParams.get("pType"));
            if (pType.length() > 1) {

                BasicDBList values = new BasicDBList();
                String[] pTypeArray = pType.split(",");
                for (String type : pTypeArray) {
                    values.add(new BasicDBObject("pType", ObjectUtils.toInteger(type)));
                }
                matchCondition.put("$or", values);
            } else {

                matchCondition.put("pType", ObjectUtils.toInteger(pType));
            }
        }
        if (queryParams.get("status") != null) {
            matchCondition.put("status", ObjectUtils.toInteger(queryParams.get("status")));
        }
        if (queryParams.get("isNotCarousel") != null) {
            matchCondition.put("isCarousel", new BasicDBObject("$ne", 1));
        }
        matchDBObj.put("$match", matchCondition);
        pipeline.add(matchDBObj);

        // 分组
        BasicDBObject groupDBObj = new BasicDBObject();
        BasicDBObject groupCondition = new BasicDBObject();
        BasicDBObject groupParamDBObj = new BasicDBObject();
        for (String groupParam : groupParams) {

            groupParamDBObj.put(groupParam, "$" + groupParam);
        }
        groupCondition.put("_id", groupParamDBObj);
        groupCondition.put(sumColumnName, new BasicDBObject("$sum", "$" + sumColumnName));
        groupDBObj.put("$group", groupCondition);
        pipeline.add(groupDBObj);

        // 获取合计值
        AggregationOutput output = this.dbCollection.aggregate(pipeline);
        Iterator<DBObject> result = output.results().iterator();

        while (result.hasNext()) {

            DBObject dbObject = result.next();
            DBObject keys = (DBObject) dbObject.get("_id");
            StringBuilder sb = new StringBuilder();

            // 用分组列的值作为key，各分组列的值以_分隔
            for (String groupParam : groupParams) {

                sb.append(ObjUtils.toString(keys.get(groupParam)));
                sb.append("_");
            }
            sb.deleteCharAt(sb.lastIndexOf("_"));
            String key = sb.toString();
            Long totalNum = ObjUtils.toLong(dbObject.get(sumColumnName), 0L);
            totalNumMap.put(key, totalNum);
        }

        return totalNumMap;
    }

    //2015.10.29

    /**
     * 根据条件查询相应内容
     *
     * @param params
     * @return
     */
    public List<Program> getAllProgram(Map<String, Object> params) {
        List<Program> result = Lists.newArrayList();
        try {
            //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
            BasicDBObject query = getBasicDBObjectByParams(params);
            DBCursor cur = this.dbCollection.find(query);
            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                result.add(program);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return result;
    }

    /**
     * 预约人数自增或自减
     *
     * @param programId 视频ID
     * @param interval  增减幅度
     * @return 预约人数
     */
    public Long modifyOrderNum(Long programId, Long interval) {

        // 预约人数
        Long orderNum = 0L;

        BasicDBObject query = new BasicDBObject();
        query.put("_id", programId);

        // 当做自减操作时，要追加当前值>0的限制条件，避免出现负值
        if (interval < 0) {

            query.put("orderNum", new BasicDBObject("$gt", 0L));
        }

        DBObject dbObj = this.dbCollection.findAndModify(query, new BasicDBObject("$inc",
                new BasicDBObject("orderNum", interval)));

        if (dbObj != null) {

            orderNum = ObjectUtils.toLong(dbObj.get("orderNum"));
        }

        return orderNum;
    }

    /**
     * 剔除isCarousel
     *
     * @param program
     */
    public void removeIsCarousel(Program program) {
        DBObject queryCondition = new BasicDBObject();
        Long pid = program.getId();
        queryCondition.put("_id", pid); // 用来确定要修改的文档
        BasicDBObject setValue = (BasicDBObject) ProgramConvert.castProgramToDBObject(program);
        //剔除isCarousel
        if (setValue.containsField("isCarousel")) {
            setValue.removeField("isCarousel");
        }
        if (setValue.containsField("_id")) {
            setValue.removeField("_id");
        }
        DBRef userRef = new DBRef(this.dbCollection.getDB(), "user", program.getUserId());
        setValue.put("user", userRef);
        DBObject updateValue = new BasicDBObject("$set", setValue);
        this.dbCollection.update(queryCondition, updateValue, true, false);
    }

    public List<Program> getAllProgramByParams(Map<String, Object> params) {
        List<Program> result = Lists.newArrayList();
        try {
            //dbObj 类似于 mysql 查询中的 where 关键字，可以添加各种搜索条件
            BasicDBObject query = getBasicDBObjectByParams(params);
            DBCursor cur = this.dbCollection.find(query);
            while (cur.hasNext()) {
                DBObject dbObject = cur.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                result.add(program);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return result;
    }

    /**
     * 解决内嵌集合里排序问题  用于视频分类管理
     *
     * @param params
     * @param orders
     * @param start
     * @param limit
     * @return
     */
    public List<Program> getProgramListByParamsUnwindCollection(Map<String, Object> params, Map<String, Object> orders, Integer start, Integer limit) {

        List<Program> programList = Lists.newArrayList();
        if (limit == null) {

            limit = DEFAULT_NUM;
        }

        try {

            List<DBObject> pipeline = Lists.newArrayList();

            // 检索条件
            BasicDBObject matchDBObj = new BasicDBObject();
            BasicDBObject matchCondition = new BasicDBObject();

            // 视频类型
            if (params.get("pType") != null) {
                String pType = ObjectUtils.toString(params.get("pType"));
                if (pType.length() > 2) {
                    BasicDBList values = new BasicDBList();
                    String[] pTypeArray = pType.split(",");
                    for (String type : pTypeArray) {
                        values.add(ObjectUtils.toInteger(type));
                    }
                    matchCondition.put("pType", new BasicDBObject("$in", values));
                } else {
                    matchCondition.put("pType", ObjectUtils.toInteger(pType));
                }
            }
            // 视频名称
            if (params.get("pName") != null) {
                BasicDBList values = new BasicDBList();
                Pattern pattern = Pattern.compile("^.*" + params.get("pName") + ".*$", Pattern.CASE_INSENSITIVE);
                values.add(new BasicDBObject("_id", ObjectUtils.toLong(params.get("pName"))));
                values.add(new BasicDBObject("pName", pattern));
                matchCondition.put("$or", values);
            }
            // 视频状态
            if (params.get("timeShift") != null) {
                BasicDBList statusList = new BasicDBList();
                statusList.add(-1);
                statusList.add(1);
                matchCondition.put("status", new BasicDBObject("$in", statusList));
            } else {
                if (params.get("status") != null) {
                    String status = ObjectUtils.toString(params.get("status"));
                    if (status.length() > 2) {
                        BasicDBList values = new BasicDBList();
                        String[] pTypeArray = status.split(",");
                        for (String type : pTypeArray) {
                            values.add(ObjectUtils.toInteger(type));
                        }
                        matchCondition.put("status", new BasicDBObject("$in", values));
                    } else {
                        matchCondition.put("status", ObjectUtils.toInteger(status));
                    }
                }
            }
            // 视频分类id
            if (params.get("assorts.id") != null) {

                matchCondition.put("assorts.id", ObjectUtils.toInteger(params.get("assorts.id")));
            }

            matchDBObj.put("$match", matchCondition);
            pipeline.add(matchDBObj);

            // 展开条件
            BasicDBObject unwindDBObj = new BasicDBObject();
            unwindDBObj.put("$unwind", "$assorts");
            pipeline.add(unwindDBObj);

            // 筛选条件
            BasicDBObject secondMatchDBObj = new BasicDBObject();
            BasicDBObject secondMatchCondition = new BasicDBObject();
            if (params.get("assorts.id") != null) {

                secondMatchCondition.put("assorts.id", ObjectUtils.toLong(params.get("assorts.id")));
            }
            secondMatchDBObj.put("$match", secondMatchCondition);
            pipeline.add(secondMatchDBObj);

            // 排序
            BasicDBObject sortDBObj = new BasicDBObject();
            BasicDBObject sortCondition = new BasicDBObject();
            if (orders.get("pType") != null)
                sortCondition.put("pType", ObjectUtils.toInteger(orders.get("pType")));
            // 优先级
            if (orders.get("assorts.priority") != null)
                sortCondition.put("assorts.priority", ObjectUtils.toLong(orders.get("assorts.priority")));
            // 创建时间
            if (orders.get("createTime") != null)
                sortCondition.put("createTime", ObjectUtils.toLong(orders.get("createTime")));
            sortDBObj.put("$sort", sortCondition);
            pipeline.add(sortDBObj);

            // 分页条件
            BasicDBObject skipDBObj = new BasicDBObject();
            skipDBObj.put("$skip", start);
            pipeline.add(skipDBObj);
            BasicDBObject limitDBObj = new BasicDBObject();
            limitDBObj.put("$limit", limit);
            pipeline.add(limitDBObj);
            AggregationOutput output = this.dbCollection.aggregate(pipeline);
            Iterator<DBObject> result = output.results().iterator();

            while (result.hasNext()) {

                DBObject dbObject = result.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                setValueToProgram(program);
                programList.add(program);
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return programList;
    }

    /**
     * 解决内嵌集合里排序问题  用于视频的活动分类管理
     *
     * @param params
     * @param orders
     * @param start
     * @param limit
     * @return
     */
    public List<Program> getProgramListByParamsUnwindCollectionForActivity(Map<String, Object> params, Map<String, Object> orders, Integer start, Integer limit) {

        List<Program> programList = Lists.newArrayList();
        if (limit == null) {

            limit = DEFAULT_NUM;
        }

        try {

            List<DBObject> pipeline = Lists.newArrayList();

            // 检索条件
            BasicDBObject matchDBObj = new BasicDBObject();
            BasicDBObject matchCondition = new BasicDBObject();

            // 视频类型
            if (params.get("pType") != null) {
                String pType = ObjectUtils.toString(params.get("pType"));
                if (pType.length() > 2) {
                    BasicDBList values = new BasicDBList();
                    String[] pTypeArray = pType.split(",");
                    for (String type : pTypeArray) {
                        values.add(ObjectUtils.toInteger(type));
                    }
                    matchCondition.put("pType", new BasicDBObject("$in", values));
                } else {
                    matchCondition.put("pType", ObjectUtils.toInteger(pType));
                }
            }
            // 视频名称
            if (params.get("pName") != null) {
                BasicDBList values = new BasicDBList();
                Pattern pattern = Pattern.compile("^.*" + params.get("pName") + ".*$", Pattern.CASE_INSENSITIVE);
                values.add(new BasicDBObject("_id", ObjectUtils.toLong(params.get("pName"))));
                values.add(new BasicDBObject("pName", pattern));
                matchCondition.put("$or", values);
            }
            // 视频状态
            if (params.get("timeShift") != null) {
                BasicDBList statusList = new BasicDBList();
                statusList.add(-1);
                statusList.add(1);
                matchCondition.put("status", new BasicDBObject("$in", statusList));
            } else {
                if (params.get("status") != null) {
                    String status = ObjectUtils.toString(params.get("status"));
                    if (status.length() > 2) {
                        BasicDBList values = new BasicDBList();
                        String[] pTypeArray = status.split(",");
                        for (String type : pTypeArray) {
                            values.add(ObjectUtils.toInteger(type));
                        }
                        matchCondition.put("status", new BasicDBObject("$in", values));
                    } else {
                        matchCondition.put("status", ObjectUtils.toInteger(status));
                    }
                }
            }
            // 活动分类id
            if (params.get("activity.id") != null) {

                matchCondition.put("activity.id", ObjectUtils.toInteger(params.get("activity.id")));
            }

            matchDBObj.put("$match", matchCondition);
            pipeline.add(matchDBObj);

            // 展开条件
            BasicDBObject unwindDBObj = new BasicDBObject();
            unwindDBObj.put("$unwind", "$activity");
            pipeline.add(unwindDBObj);

            // 筛选条件
            BasicDBObject secondMatchDBObj = new BasicDBObject();
            BasicDBObject secondMatchCondition = new BasicDBObject();
            if (params.get("activity.id") != null) {

                secondMatchCondition.put("activity.id", ObjectUtils.toLong(params.get("activity.id")));
            }
            secondMatchDBObj.put("$match", secondMatchCondition);
            pipeline.add(secondMatchDBObj);

            // 排序
            BasicDBObject sortDBObj = new BasicDBObject();
            BasicDBObject sortCondition = new BasicDBObject();
            if (orders.get("pType") != null)
                sortCondition.put("pType", ObjectUtils.toInteger(orders.get("pType")));
            // 优先级
            if (orders.get("activity.priority") != null)
                sortCondition.put("activity.priority", ObjectUtils.toLong(orders.get("activity.priority")));
            // 创建时间
            if (orders.get("createTime") != null)
                sortCondition.put("createTime", ObjectUtils.toLong(orders.get("createTime")));
            sortDBObj.put("$sort", sortCondition);
            pipeline.add(sortDBObj);

            // 分页条件
            BasicDBObject skipDBObj = new BasicDBObject();
            skipDBObj.put("$skip", start);
            pipeline.add(skipDBObj);
            BasicDBObject limitDBObj = new BasicDBObject();
            limitDBObj.put("$limit", limit);
            pipeline.add(limitDBObj);
            AggregationOutput output = this.dbCollection.aggregate(pipeline);
            Iterator<DBObject> result = output.results().iterator();

            while (result.hasNext()) {

                DBObject dbObject = result.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                setValueToProgram(program);
                programList.add(program);
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return programList;
    }

    /**
     * 根据指定条件合计用户在一定时期内的点赞总数
     *
     * @param key     分组key
     * @param cond    查询条件
     * @param initial 初始值
     * @param reduce  计算函数
     * @return
     */
    public DBObject getTotalLikeNumInPeriodByUser(DBObject key, DBObject cond, DBObject initial, String reduce) {
        DBObject result = this.dbCollection.group(key, cond, initial, reduce);
        return result;
    }

    /**
     * 获取指定标签视频列表
     *
     * @param query
     * @param order
     * @param start
     * @param limit
     * @return
     */
    public List<Program> getProgramListByTag(Map<String, Object> query, Map<String, Object> order, Integer start, Integer limit) {

        List<Program> programList = Lists.newArrayList();
        if (limit == null) {

            limit = DEFAULT_NUM;
        }
        try {
            List<DBObject> pipeline = Lists.newArrayList();
            // 检索条件
            BasicDBObject matchDBObj = new BasicDBObject();
            matchDBObj.put("$match", query);
            pipeline.add(matchDBObj);

            // 展开条件
            BasicDBObject unwindDBObj = new BasicDBObject();
            unwindDBObj.put("$unwind", "$assorts");
            pipeline.add(unwindDBObj);

            // 筛选条件
            BasicDBObject secondMatchDBObj = new BasicDBObject();
            BasicDBObject secondMatchCondition = new BasicDBObject();
            if (query.get("assorts.id") != null) {

                secondMatchCondition.put("assorts.id", ObjectUtils.toLong(query.get("assorts.id")));
            }
            secondMatchDBObj.put("$match", secondMatchCondition);
            pipeline.add(secondMatchDBObj);

            // 排序
            BasicDBObject sortDBObj = new BasicDBObject();
            sortDBObj.put("$sort", order);
            pipeline.add(sortDBObj);

            // 分页条件
            BasicDBObject skipDBObj = new BasicDBObject();
            skipDBObj.put("$skip", (start - 1) * limit);
            pipeline.add(skipDBObj);
            BasicDBObject limitDBObj = new BasicDBObject();
            limitDBObj.put("$limit", limit);
            pipeline.add(limitDBObj);
            AggregationOutput output = this.dbCollection.aggregate(pipeline);
            Iterator<DBObject> result = output.results().iterator();

            while (result.hasNext()) {

                DBObject dbObject = result.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                programList.add(program);
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return programList;
    }

    public List<Program> getLiveProgram(Integer start, Integer limit, Map<String, Object> params) {
        DBObject order = new BasicDBObject();
        order.put("isCarousel", 1);
        order.put("priority", -1);
        order.put("createTime", -1);
        List<DBObject> programList = selectAll(start, limit, params, order);
        List<Program> result = Lists.newArrayList();
        for (DBObject obj : programList) {

            Program program = ProgramConvert.castDBObjectToProgram(obj);

            // 设定视频参与信息
            setInvolInfor(program);

            result.add(program);
        }
        return result;
    }

    public List<Program> myVideoList(Long userId, Integer start, Integer limit, Long timestamp) {
        DBObject query = new BasicDBObject("userId", userId);
        query.put("sourceType", Constants.SOURCE_TYPE_LE_UPLOAD_VIDEO);
        DBObject order = new BasicDBObject("createTime", -1);
        if (start != 1) {
            if (timestamp != null && timestamp > 0) {
                query.put("createTime", new BasicDBObject("$lt", timestamp));
            }
        }
        DBCursor cur = dbCollection.find(query).sort(order).skip((start - 1) * limit).limit(limit);
        List<Program> programList = Lists.newArrayList();
        while (cur.hasNext()) {
            DBObject dbObject = cur.next();
            programList.add(ProgramConvert.castDBObjectToProgram(dbObject));
        }
        return programList;
    }

    /**
     * 获取指定活动下的视频列表
     *
     * @param query 查询条件
     * @param order 排序字段
     * @param start 开始页号
     * @param limit 每页长度
     * @return
     */
    public List<Program> getProgramListByAct(Map<String, Object> query, Map<String, Object> order, Integer start, Integer limit) {

        List<Program> programList = Lists.newArrayList();
        if (limit == null) {

            limit = DEFAULT_NUM;
        }
        try {
            List<DBObject> pipeline = Lists.newArrayList();
            // 检索条件
            BasicDBObject matchDBObj = new BasicDBObject();
            matchDBObj.put("$match", query);
            pipeline.add(matchDBObj);

            // 展开条件
            BasicDBObject unwindDBObj = new BasicDBObject();
            unwindDBObj.put("$unwind", "$activity");
            pipeline.add(unwindDBObj);

            // 筛选条件
            BasicDBObject secondMatchDBObj = new BasicDBObject();
            BasicDBObject secondMatchCondition = new BasicDBObject();
            if (query.get("activity.id") != null) {

                secondMatchCondition.put("activity.id", ObjectUtils.toLong(query.get("activity.id")));
            }
            secondMatchDBObj.put("$match", secondMatchCondition);
            pipeline.add(secondMatchDBObj);

            // 排序
            BasicDBObject sortDBObj = new BasicDBObject();
            sortDBObj.put("$sort", order);
            pipeline.add(sortDBObj);

            // 分页条件
            BasicDBObject skipDBObj = new BasicDBObject();
            skipDBObj.put("$skip", (start - 1) * limit);
            pipeline.add(skipDBObj);
            BasicDBObject limitDBObj = new BasicDBObject();
            limitDBObj.put("$limit", limit);
            pipeline.add(limitDBObj);
            AggregationOutput output = this.dbCollection.aggregate(pipeline);
            Iterator<DBObject> result = output.results().iterator();

            while (result.hasNext()) {

                DBObject dbObject = result.next();
                Program program = ProgramConvert.castDBObjectToProgram(dbObject);
                programList.add(program);
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return programList;
    }

    /**
     * 设定视频参与信息
     *
     * @param program
     */
    public void setInvolInfor(Program program) {

        // 当为直播时使用在线数据
        if (program.getPType() == 1) {

            // 点赞数
            Long baseLikeNum = ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().hget(CacheConstants.PROGRAM_BASE_LIKENUM, ObjectUtils.toString(program.getId())), 0L);
            Long likeNum = ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().get(Constants.LIVE_ONLINE_LIKE_KEY + program.getId()), 0L);
            program.setLikeNum(baseLikeNum + likeNum);

            // 评论数
            program.setCommentNum(ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().get(Constants.LIVE_ONLINE_COMMENT_KEY + program.getId()), 0l));

            Long baseNum = ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().hget(CacheConstants.PROGRAM_BASE_WATCHNUM, ObjectUtils.toString(program.getId())), 0L);

            // 观看人数
            Long watchNum = ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().zcard(Constants.LIVE_ONLINE_USER_LIST_KEY + program.getId()), 0L);
            program.setWatchNum(baseNum + watchNum);

            // 观看人次
            Long watchCount = ObjectUtils.toLong(jedisDAO.getJedisReadTemplate().get(Constants.LIVE_ONLINE_WATCH_KEY + program.getId()), 0L);
            program.setWatchCount(baseNum + watchCount);
        }
    }
}
